package cn.nn200433.mqtt.service.impl;

import cn.hutool.core.thread.ThreadUtil;
import cn.hutool.core.util.StrUtil;
import cn.nn200433.mqtt.config.MqttClientProperties;
import cn.nn200433.mqtt.service.IMqttConnectService;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttAsyncClient;
import org.eclipse.paho.client.mqttv3.MqttCallbackExtended;
import org.eclipse.paho.client.mqttv3.MqttMessage;

import java.util.List;

/**
 * 默认的mqtt回调服务实现
 *
 * @author nn200433
 * @date 2021-07-26 10:43:32
 */
@Slf4j
public class DefaultMqttCallbackServiceImpl implements MqttCallbackExtended {

    /** mqtt客户端 */
    private MqttAsyncClient      client;
    /** mqtt处理上下文 */
    private MqttHandlerContext   mqttHandlerContext;
    /** mqtt配置信息 */
    private MqttClientProperties mqttConfig;

    public DefaultMqttCallbackServiceImpl(MqttAsyncClient client, MqttHandlerContext mqttHandlerContext,
                                          MqttClientProperties mqttConfig) {
        this.client             = client;
        this.mqttHandlerContext = mqttHandlerContext;
        this.mqttConfig         = mqttConfig;
    }

    @Override
    public void connectComplete(boolean reconnect, String serverURI) {
        final Boolean autoSubscribeDefault = mqttConfig.getAutoSubscribeDefault();
        final String  subTopics            = mqttConfig.getSubTopics();
        try {
            // 订阅默认主题
            if (autoSubscribeDefault) {
                final List<String> subTopicList = StrUtil.split(subTopics, StrUtil.COMMA, Boolean.TRUE, Boolean.TRUE);
                for (String subTopic : subTopicList) {
                    log.info("--->【MQTT】客户端成功建立连接，开始订阅主题 {} ...", subTopic);
                    client.subscribe(subTopic, mqttConfig.getSubTopicQos());
                }
            }

            // 异步进行一些额外操作，如记录消息日志
            final String connectMsg = reconnect ? "客户端成功建立重连..." : "客户端成功建立连接...";
            ThreadUtil.execute(() -> getConnectService().connectComplete(client, mqttConfig, reconnect, serverURI, connectMsg));
        } catch (Exception e) {
            log.error("--->【MQTT】重新订阅主题 {} 失败...", subTopics, e);
        }
    }

    @Override
    public void connectionLost(Throwable cause) {
        ThreadUtil.execute(() -> reConnect(cause, 10));
    }

    @Override
    public void messageArrived(String topic, MqttMessage message)
            throws Exception {
        try {
            mqttHandlerContext.getHandlerStrategy(topic).subscribeMsgArrivedHandle(mqttConfig, topic, message);
        } catch (Exception e) {
            log.error("--->【MQTT】获取 {} 策略处理数据异常，请检查是否定义该策略...", topic);
            log.error("--->【MQTT】可以实现 IMqttHandlerService 接口进行定义消息处理策略...");
            log.error("", e);
        }
    }

    @Override
    public void deliveryComplete(IMqttDeliveryToken token) {
        log.debug("--->【MQTT】deliveryComplete：{}", token.isComplete());
    }

    /**
     * 重新连接
     *
     * @param cause   导致
     * @param tryWait 每次重连后等待时长，单位：秒
     * @author nn200433
     */
    private void reConnect(Throwable cause, int tryWait) {
        long reconnectTimes = 1;
        while (true) {
            try {
                if (client.isConnected()) {
                    return;
                }

                // 异步进行一些额外操作，如记录消息日志
                final String warningStr = StrUtil.format("客户端正尝试第 {} 次重新连...", reconnectTimes++);
                ThreadUtil.execute(() -> getConnectService().connectionLost(client, mqttConfig, cause, warningStr));

                // 用于触发尝试重新连接
                client.reconnect();
            } catch (Exception e) {
                log.error("--->【MQTT】客户端重连失败...", e);
            }
            ThreadUtil.sleep(tryWait * 1000);
        }
    }

    /**
     * 获取连接服务
     *
     * @return {@link IMqttConnectService }
     * @author nn200433
     */
    private IMqttConnectService getConnectService() {
        return mqttHandlerContext.getConnectService();
    }

}
